package com.sjk.flink.kafka;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.sjk.flink.pojo.Performance;
import com.sjk.flink.util.ObjectMapperUtil;
import org.apache.flink.streaming.connectors.kafka.KafkaSerializationSchema;
import org.apache.kafka.clients.producer.ProducerRecord;
import javax.annotation.Nullable;

public class PerformanceKafkaSerializationSchema implements KafkaSerializationSchema<Performance> {

    private static final long serialVersionUID = 8497940668660042203L;
    public PerformanceKafkaSerializationSchema() {
    }

    @Override
    public ProducerRecord<byte[], byte[]> serialize(Performance performance, @Nullable Long aLong) {
        String json = "json转换错误";
        if(performance!=null) {
            try {
                ObjectMapper objectMapper= ObjectMapperUtil.getObjectMapper();
                json = objectMapper.writeValueAsString(performance);
            } catch (JsonProcessingException e) {
                e.printStackTrace();
            }
        }
        return new ProducerRecord<byte[], byte[]>("FLINK_XNCS_SJ", json.getBytes());
    }
}